package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: guoml
 * @Description: 针对企查查t_ipo_company_important_wgcl表和聚源LC_Deregulation表进行数据映射开发
 * @Date: 2021/12/7 12:51
 */

public class TIpoCompanyImportantWgcl {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_ipo_company_important_wgcl").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        spark.sql("select * from a_dc_ep_ods.CT_SystemConst m WHERE m.LB in (1165,1163,1164,1015) and m.DM is not null").createOrReplaceTempView("first_table");
//        spark.catalog().cacheTable("first_table");
        spark.sql("select a.*,b.SecuCode from a_dc_ep_ods.LC_Deregulation a inner join (select SecuCode,CompanyCode from a_dc_ep_ods.SecuMain where ListedState = 1 and SecuCategory = 1) b on a.CompanyCode = b.CompanyCode ").createOrReplaceTempView("tmp_table1");
        spark.sql("select a.*,c.ms as ms1 from tmp_table1 a left join (SELECT ms,dm FROM first_table m WHERE m.LB = 1165) c ON a.ReasonType = c.DM ").createOrReplaceTempView("tmp_table2 ");
//        spark.catalog().dropTempView("tmp_table1");
        spark.sql("select a.*,SE.code from tmp_table2 a left join (select j.* from a_dc_ep_ods.LC_Deregulation_SE j where j.TypeCode = 1 and j.id is not null) SE on a.ID = SE.ID ").createOrReplaceTempView("tmp_table3");
//        spark.catalog().dropTempView("tmp_table2");
        spark.sql("select a.*,d.ms as ms2 from tmp_table3 a left join (select * from first_table m WHERE m.LB = 1163) d on a.Code = d.DM ").createOrReplaceTempView("tmp_table4");
//        spark.catalog().dropTempView("tmp_table3");
        spark.sql("select a.*,e.ms as ms3 from tmp_table4 a left join (SELECT * FROM first_table m WHERE m.LB = 1164) e ON a.AdminType = e.DM ").createOrReplaceTempView("tmp_table5");
//        spark.catalog().dropTempView("tmp_table4");
        spark.sql("select a.*,f.ms as ms4 from tmp_table5 a left join (SELECT * FROM first_table m WHERE m.LB = 1015) f ON a.AdminInst = f.DM ").createOrReplaceTempView("tmp_table6");
//        spark.catalog().dropTempView("tmp_table5");
//        spark.catalog().dropTempView("first_table");
        //获取中数增量数据表中的新增数据（left join）
        Dataset<Row> insert_data = spark.sql("SELECT md5(a.ID),\n" +
                "       a.SecuCode,\n" +
                "       a.InfoPublDate,\n" +
                "       a.MS1,\n" +
                "       a.MS2,\n" +
                "       a.EventContent,\n" +
                "       a.MS3,\n" +
                "       a.Penalty,\n" +
                "       a.MS4,\n" +
                "       a.ComPenalSum,\n" +
                "       '' as created_date,\n" +
                "       '' as updated_date,\n" +
                "       '' as encrypt_key,\n" +
                "       regexp_replace(current_date(),'-',''),\n" +
                "       \"1\" as isadd\n" +
                "FROM\n" +
                "  tmp_table6 a");
        insert_data.createOrReplaceTempView("tmp_t_ipo_company_important_wgcl1");
        spark.sql("insert overwrite table a_dc_ep_incr.t_ipo_company_important_wgcl select * from tmp_t_ipo_company_important_wgcl1");
//        spark.catalog().dropTempView("tmp_table6");
//        spark.catalog().dropTempView("tmp_t_ipo_company_important_wgcl1");
        spark.stop();
    }
}
